package hash

import (
	"fmt"
	"hash/crc32"
	"sort"
	"strconv"
	"sync"

	"github.com/meilihao/golib/v2/log"
)

var (
	DefaultHashFn = func(node string) uint32 {
		return crc32.ChecksumIEEE([]byte(node))
	}
)

// [golang 实现一致性 hash 算法](https://xie.infoq.cn/article/78043810ecc807d1896c6f3f2)
// [一致性hash算法原理及golang实现](https://segmentfault.com/a/1190000013533592)
type ConsistentHash struct {
	sync.RWMutex
	hashFn           func(string) uint32
	numberOfReplicas int // 虚拟结点数
	hashSortedNodes  []uint32
	circle           map[uint32]string
	Nodes            map[string]struct{}
}

func NewConsistentHash(fn func(string) uint32, numberOfReplicas int, nodes []string) *ConsistentHash {
	c := &ConsistentHash{
		hashFn: fn, numberOfReplicas: numberOfReplicas,
		hashSortedNodes: make([]uint32, 0),
		circle:          map[uint32]string{},
		Nodes:           map[string]struct{}{},
	}

	for _, node := range nodes {
		c.Add(node)
	}

	return c
}

func (c *ConsistentHash) Add(node string) {
	c.Lock()
	defer c.Unlock()

	if _, ok := c.Nodes[node]; ok {
		return
	}
	c.Nodes[node] = struct{}{}

	var virtualKey uint32
	for i := 0; i < c.numberOfReplicas; i++ {
		virtualKey = c.hashFn(node + strconv.Itoa(i))
		c.circle[virtualKey] = node
		c.hashSortedNodes = append(c.hashSortedNodes, virtualKey)
	}

	sort.Slice(c.hashSortedNodes, func(i, j int) bool {
		return c.hashSortedNodes[i] < c.hashSortedNodes[j]
	})

	log.Glog.Debug(fmt.Sprintf("after adding, consistent hash circle has management nodes%v, %d virtual nodes now",
		c.Nodes,
		len(c.circle),
	))
}

func (c *ConsistentHash) Remove(node string) {
	c.Lock()
	defer c.Unlock()

	if _, ok := c.Nodes[node]; !ok {
		return
	}

	delete(c.Nodes, node)
	c.Nodes[node] = struct{}{}

	log.Glog.Debug(fmt.Sprintf("the consistent hash ring currently has nodes%v", c.Nodes))

	var virtualKey uint32
	for i := 0; i < c.numberOfReplicas; i++ {
		virtualKey = c.hashFn(node + strconv.Itoa(i))
		delete(c.circle, virtualKey)
	}

	c.hashSortedNodes = nil
	for node = range c.Nodes {
		for i := 0; i < c.numberOfReplicas; i++ {
			virtualKey = c.hashFn(node + strconv.Itoa(i))
			c.hashSortedNodes = append(c.hashSortedNodes, virtualKey)
		}
	}

	sort.Slice(c.hashSortedNodes, func(i, j int) bool {
		return c.hashSortedNodes[i] < c.hashSortedNodes[j]
	})

	log.Glog.Debug(fmt.Sprintf("after removing, consistent hash circle has management nodes%v, %d virtual nodes now",
		c.Nodes,
		len(c.circle),
	))
}

func (c *ConsistentHash) HasNode(node string) bool {
	c.Lock()
	defer c.Unlock()

	var ok bool
	var virtualKey uint32

	for i := 0; i < c.numberOfReplicas; i++ {
		virtualKey = c.hashFn(node + strconv.Itoa(i))
		if _, ok = c.circle[virtualKey]; ok {
			return true
		}
	}

	return false
}

func (c *ConsistentHash) Get(node string) string {
	c.RLock()
	defer c.RUnlock()

	if len(c.circle) == 0 {
		return ""
	}

	var ok bool
	var virtualKey uint32
	var value string

	virtualKey = c.hashFn(node)

	value, ok = c.circle[virtualKey]
	if !ok {
		i := sort.Search(len(c.hashSortedNodes), func(i int) bool { return c.hashSortedNodes[i] >= virtualKey })

		if i == len(c.hashSortedNodes) {
			i = 0
		}

		return c.circle[c.hashSortedNodes[i]]
	}

	return value
}
